package com.platform.apex.app;

import org.apache.hadoop.conf.Configuration;

import com.datatorrent.api.DAG;
import com.datatorrent.api.StreamingApplication;
import com.datatorrent.api.annotation.ApplicationAnnotation;
import com.datatorrent.lib.io.ConsoleOutputOperator;
import com.platform.apex.input.HDFSInoutOperator;
import com.platform.apex.output.ElasticAppSearchConnectable;
import com.platform.apex.output.ElasticAppSearchMapOutputOperator;

@ApplicationAnnotation(name = "HDFSToESApplicaton")
public class StaticToESApplicaton implements StreamingApplication{

	@Override
	public void populateDAG(DAG dag, Configuration conf) {
		loadDag(dag, conf);
		
	}

	
	@SuppressWarnings({ "unchecked", "rawtypes" })
	private void loadDag(DAG dag, Configuration conf) {
		HDFSInoutOperator hi = dag.addOperator("hi", HDFSInoutOperator.class);
		ElasticAppSearchMapOutputOperator output = dag.addOperator("elasticSearchWriter", new ElasticAppSearchMapOutputOperator());
		ElasticAppSearchConnectable store = new ElasticAppSearchConnectable();
		store.setHostName(conf.get("dt.operator.elasticSearchWriter.prop.hostName"));
		store.setPort(conf.getInt("dt.operator.elasticSearchWriter.prop.port", 9300));
		store.setClusterName(conf.get("dt.operator.elasticSearchWriter.prop.clusterName"));
		output.setStore(store);
//		output.setIndexName(conf.get("dt.operator.elasticSearchWriter.prop.indexName"));
//		output.setType(conf.get("dt.operator.elasticSearchWriter.prop.type"));
		
//		ConsoleOutputOperator cons = dag.addOperator("console", new ConsoleOutputOperator());
//		
//		dag.addStream("ht-to-con", hi.esOutPut, cons.input);
		dag.addStream("static-to-es", hi.esOutPut, output.input);
		
	}
}
